package com.lry.registy.zookeeper;

import com.lry.common.CommonConstants;
import com.lry.event.*;
import com.lry.registy.AbstractRegister;
import com.lry.registy.RegistryService;
import com.lry.registy.URL;
import com.lry.spi.LoadLevel;
import com.lry.util.CollectionUtils;
import com.lry.util.StringUtils;

import java.util.ArrayList;
import java.util.List;

import static com.lry.common.CommonConstants.*;

@LoadLevel(name = "zookeeper")
public class ZookeeperRegister extends AbstractRegister implements RegistryService {

    private AbstractZookeeperClient zkClient;

    private String root = SLASH + ROOT;


    public ZookeeperRegister(String address) {
        this.zkClient = new CuratorZookeeperClient(address);
    }


    @Override
    public List<String> getProviderIps(String serviceName,String implName) {
        List<String> ipList = this.zkClient.getChildrenData(URL.getParentProviderPath(serviceName,implName));
        return ipList;
    }

    @Override
    public String getProviderContent(URL url){
        return getProviderContent(url.getProviderPath());
    }

    private String getProviderContent(String providerPath){
        return this.zkClient.getNodeData(providerPath);
    }

    @Override
    public void register(URL url) {
        if (!this.zkClient.existNode(root)) {
            zkClient.createPersistentData(root, "");
        }
        String nodeContent = url.buildProviderNodeContent();
        String providerPath = url.getProviderPath();
        if (zkClient.existNode(providerPath)) {
            zkClient.deleteNode(providerPath);
        }
        zkClient.createTemporaryData(providerPath, nodeContent);
        super.register(url);
    }

    @Override
    public void unRegister(URL url) {
        zkClient.deleteNode(url.getProviderPath());
        super.unRegister(url);
    }

    @Override
    public void subscribe(URL url) {
        if (!this.zkClient.existNode(root)) {
            zkClient.createPersistentData(root, "");
        }
        String nodeContent = url.buildConsumerNodeContent();
        String consumerPath = url.getConsumerPath();
        if (zkClient.existNode(consumerPath)) {
            zkClient.deleteNode(consumerPath);
        }

        zkClient.createTemporarySeqData(consumerPath, nodeContent);
        super.subscribe(url);
    }

    @Override
    public void unSubscribe(URL url) {
        this.zkClient.deleteNode(url.getConsumerPath());
        super.unSubscribe(url);
    }

    @Override
    public void doAfterSubscribe(URL url) {
        //监听是否有新的服务注册
        String parProviderPath = URL.getParentProviderPath(url.getServiceName(),url.getImplName());
        watchChildNode(parProviderPath);

        List<String> ips = (List<String>) url.getParam().get(CommonConstants.IPS);
        for (String ip : ips) {
            watchNodeData(parProviderPath+SLASH+ip);
        }
    }

    public void watchNodeData(String providerPath){
        zkClient.watchNodeData(providerPath, watchedEvent-> {
            String nodeData = zkClient.getNodeData(providerPath);
            if(StringUtils.isNotBlank(nodeData)){
                ProviderNodeWrapper providerNodeWrapper = URL.buildProviderNodeFromContent(nodeData);
                RpcListenerLoader.sendEvent(new NodeChangeEvent(providerNodeWrapper));
            }

            //收到回调之后在注册一次监听，这样能保证一直都收到消息
            watchNodeData(providerPath);
        });
    }

    public void watchChildNode(String parProviderPath){
        zkClient.watchChildNodeData(parProviderPath, watchedEvent-> {
                List<String> childrenDataList = zkClient.getChildrenData(parProviderPath);

                List<ProviderNodeWrapper> nodes = new ArrayList<>();

                //weight
                for (String ips : childrenDataList) {
                    String providerContent = getProviderContent(parProviderPath + ips);
                    if(StringUtils.isBlank(providerContent)){
                        continue;
                    }
                    ProviderNodeWrapper providerNodeWrapper = URL.buildProviderNodeFromContent(providerContent);
                    nodes.add(providerNodeWrapper);
                }

                if(CollectionUtils.isNotEmpty(nodes)){
                    String[] names = URL.getServiceNameFromParentProviderPath(parProviderPath);
                    ProviderChangeWrapper providerChangeWrapper = new ProviderChangeWrapper(names[0],names[1], nodes);
                    RpcListenerLoader.sendEvent(new ProviderChangeEvent(providerChangeWrapper));
                }
                //收到回调之后在注册一次监听，这样能保证一直都收到消息
                watchChildNode(parProviderPath);
        });
    }

    @Override
    public void doBeforeSubscribe(URL url) {

    }



}
